package com.darrenchan.spark

import java.io.PrintWriter
import java.net.ServerSocket

import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import org.apache.log4j.{Logger, Level}
import java.sql.Timestamp

import scala.util.Random

/**
  * Created by tsun on 19/4/26.
  */
object StructuredStreaming {
    Logger.getLogger("org").setLevel(Level.ERROR)
    def main(args: Array[String]) {
        val spark = SparkSession
                .builder
                .master("local[2]")
                .appName("StructuredNetworkWordCount")
                .getOrCreate()

        import spark.implicits._

        val lines = spark.readStream
                .format("socket")
                .option("host", "localhost")
                .option("port", 9999)
                .load()

        // Split the lines into words
        val words = lines.as[String].flatMap(_.split(" "))

        // Generate running word count
        val wordCounts = words.groupBy("value").count()

        val query = wordCounts.writeStream
//                .outputMode("update")
                .outputMode("complete")
                .format("console")
                .start()

        query.awaitTermination()
    }
}


object StructProducerWithTimestamp {
    Logger.getLogger("org").setLevel(Level.ERROR)

    def main(args: Array[String]) {

        val random = new Random()

        // Maximum number of events per second
        val MaxEvents = 6

        // Read the list of possible names
        val namesResource = this.getClass.getResourceAsStream("/names.csv")
        val names = scala.io.Source.fromInputStream(namesResource)
                .getLines()
                .toList
                .head
                .split(",")
                .toSeq

        // Generate a sequence of possible products
        val products = Seq(
            "iPhone Cover" -> 9.99,
            "Headphones" -> 5.49,
            "Samsung Galaxy Cover" -> 8.95,
            "iPad Cover" -> 7.49
        )

        /** Generate a number of random product events */
        def generateProductEvents(n: Int) = {
            (1 to n).map { i =>
                val (product, price) = products(random.nextInt(products.size))
                val user = random.shuffle(names).head
                (user, product, price, System.currentTimeMillis())
            }
        }

        // create a network producer
        val listener = new ServerSocket(9999)
        println("Listening on port: 9999")

        while (true) {
            val socket = listener.accept()
            new Thread() {
                override def run = {
                    println("Got client connected from: " + socket.getInetAddress)
                    val out = new PrintWriter(socket.getOutputStream(), true)

                    while (true) {
                        Thread.sleep(1000)
                        val num = random.nextInt(MaxEvents)
                        val productEvents = generateProductEvents(num)
                        productEvents.foreach{ event =>
                            out.write(event.productIterator.mkString(","))
                            out.write("\n")
                        }
                        out.flush()
                        println(s"Created $num events...")
                    }
                    socket.close()
                }
            }.start()
        }
    }
}


object StructuredNetworkWordCountWindowed {
    Logger.getLogger("org").setLevel(Level.ERROR)

    def main(args: Array[String]) {

        val windowDuration = "10 seconds"
        val slideDuration = "5 seconds"

        val spark = SparkSession
                .builder
                .master("local[2]")
                .appName("StructuredNetworkWordCount")
                .getOrCreate()

        import spark.implicits._

        val lines = spark.readStream
                .format("socket")
                .option("host", "localhost")
                .option("port", 9999)
                //      .option("includeTimestamp", true)
                .load()

        // Split the lines into words, retaining timestamps
        val words = lines.as[String].map(line => {
            //    val words = lines.as[(String, Timestamp)].flatMap(line =>

            val cols = line.split(",")
            (cols(0), new Timestamp(cols(3).toLong))
        }).toDF("word", "timestamp")

        // Group the data by window and word and compute the count of each group
        val windowedCounts = words
                .withWatermark("timestamp", "10 seconds")
                .groupBy(
                    window($"timestamp", windowDuration, slideDuration), $"word"
                ).count()
        //      .orderBy("window")

        // Start running the query that prints the windowed word counts to the console
        val query = windowedCounts.writeStream
                //      .outputMode("complete")
                //      .outputMode("update")
                .format("console")
                .option("truncate", "false")
                .start()

        query.awaitTermination()
    }
}

